package com.cml.infrastructure;

import com.cml.common.constant.MessageType;
import com.cml.domain.entity.DefaultMessage;
import com.cml.domain.entity.Message;
import org.apache.log4j.Logger;

import java.nio.ByteBuffer;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * 姓名：陈茂林
 * 时间：2023/10/9 13:24
 * 描述：
 */
public class MessageQueueManager {
    private static Map<Integer, BlockingQueue> SEND_QUEUES = new ConcurrentHashMap();

    private static BlockingQueue<Message> RECEIVE_QUEUE = new LinkedBlockingQueue();

    private static BlockingQueue<Message> VOTE_QUEUE = new LinkedBlockingQueue();

    private static BlockingQueue<Message> SOLT_METADATA_QUEUE = new LinkedBlockingQueue();

    private static BlockingQueue<Message> CUR_SERVER_NODE_SOLT_METADATA_QUEUE = new LinkedBlockingQueue();

    private static BlockingQueue<Message> CUR_SERVER_NODE_SOLT_METADATA_REPLICAS_QUEUE = new LinkedBlockingQueue();

    private static BlockingQueue<Message> SERVERNODE_REPLICAS_RELATIONSHIP_QUEUE = new LinkedBlockingQueue();

    private static BlockingQueue<Message> SERVERNODE_REPLICAS_SYN_QUEUE = new LinkedBlockingQueue();

    private static BlockingQueue<Message> SERVERNODE_REBALANCE_QUEUE = new LinkedBlockingQueue();

    Logger logger = Logger.getLogger(MessageQueueManager.class);

   static  {
        new ServerMessageDistribute().start();
    }
    public static BlockingQueue<Message> getReceiveQueue(){
        return MessageQueueManager.RECEIVE_QUEUE;
    }

    public static BlockingQueue createSendQueue(Integer id){
        LinkedBlockingQueue<ByteBuffer> sendQueue =
                new LinkedBlockingQueue<ByteBuffer>();
        MessageQueueManager.SEND_QUEUES.put(id, sendQueue);
        return  sendQueue;
    }

    public static void removeSendQueue(Integer id){
        MessageQueueManager.SEND_QUEUES.remove(id);
    }

    public  static Message takeVoteMessage() throws InterruptedException {
       return MessageQueueManager.VOTE_QUEUE.take();
    }

    public  static Message takeSoltMetadataMessage() throws InterruptedException {
        return MessageQueueManager.SOLT_METADATA_QUEUE.take();
    }

    public  static Message takeCurServerNodeSoltMetadataReplicasMessage() throws InterruptedException {
        return MessageQueueManager.CUR_SERVER_NODE_SOLT_METADATA_REPLICAS_QUEUE.take();
    }

    public  static Message takeServerNodeReplicasRelationshipMessage() throws InterruptedException {
        return MessageQueueManager.SERVERNODE_REPLICAS_RELATIONSHIP_QUEUE.take();
    }






    public  static Message takeCurServerNodeSoltMetadataMessage() throws InterruptedException {
        return MessageQueueManager.CUR_SERVER_NODE_SOLT_METADATA_QUEUE.take();
    }

    public static void  sendMessage(Integer nodeId,Message message) throws Exception {
       try {
           SEND_QUEUES.get(nodeId).put(message);
       }catch (Exception e){
           e.printStackTrace();
       }
    }

    public  static Message takeCurServerNodeServiceInsReplicasSYNMessage() throws InterruptedException {
        return MessageQueueManager.SERVERNODE_REPLICAS_SYN_QUEUE.take();
    }

    public static Message takeRebalanceMessage() throws InterruptedException {
        return MessageQueueManager.SERVERNODE_REBALANCE_QUEUE.take();
    }


    /**
     * 分发各类消息
     */
    private static class ServerMessageDistribute extends  Thread{
        Logger logger = Logger.getLogger(ServerMessageDistribute.class);
        @Override
        public void run() {
            logger.info("ServerMessage消息分发开始启动!");
            while (true){
                try {
                    Message message = MessageQueueManager.getReceiveQueue().take();
                    if(MessageType.VOTE == message.getMessageType()){
                        MessageQueueManager.VOTE_QUEUE.put(message);
                    }else if(MessageType.DEFAULT_MESSAGE == message.getMessageType()){
                        DefaultMessage defaultMessage = (DefaultMessage) message;
                        if(MessageType.SOLT_METADATA_MESSAGE == defaultMessage.getDefaultMessageType()){
                            MessageQueueManager.SOLT_METADATA_QUEUE.put(message);
                        }else if(MessageType.CUR_SERVER_NODE_SOLT_METADATA_MESSAGE == defaultMessage.getDefaultMessageType()){
                            CUR_SERVER_NODE_SOLT_METADATA_QUEUE.put(defaultMessage);
                        }else if(MessageType.CUR_SERVER_NODE_SOLT_METADATA_REPLICAS_MESSAGE == defaultMessage.getDefaultMessageType()){
                            MessageQueueManager.CUR_SERVER_NODE_SOLT_METADATA_REPLICAS_QUEUE.put(defaultMessage);
                        }else if(MessageType.SOLT_METADATA_REPLICAS_MESSAGE  == defaultMessage.getDefaultMessageType()){
                            MessageQueueManager.SOLT_METADATA_QUEUE.put(message);
                        }else if(MessageType.SERVERNODE_REPLICAS_RELATIONSHIP_MESSAGE== defaultMessage.getDefaultMessageType()){
                            SERVERNODE_REPLICAS_RELATIONSHIP_QUEUE.put(message);
                        } else if (MessageType.CLIENT_REQUEST_SERVICE_HEARBEAT_MESSAGE == defaultMessage.getDefaultMessageType()) {
                            SERVERNODE_REPLICAS_SYN_QUEUE.put(message);
                        }else if (MessageType.CLIENT_REPONSE_SERVICE_REGISTER_MESSAGE == defaultMessage.getDefaultMessageType()) {
                            SERVERNODE_REPLICAS_SYN_QUEUE.put(message);
                        }else if(MessageType.SERVERNODE_REBALANCE_MESSAGE == defaultMessage.getDefaultMessageType()){
                            SERVERNODE_REBALANCE_QUEUE.put(message);
                        }

                    }
                } catch (InterruptedException e) {
                   logger.error(e);
                }
            }

        }
    }
}
